package czopyk.model.analysis;

import org.apache.log4j.Logger;

import czopyk.model.analysis.Analyzer;
import czopyk.model.analysis.conditions.Condition;
import czopyk.model.analysis.conditions.Condition.StatsType;
import czopyk.model.analysis.conditions.Condition.WindowType;
import czopyk.model.analysis.conditions.LoadCondition;
import czopyk.model.analysis.conditions.ResponseTimeCondition;
import czopyk.model.analysis.conditions.ResponseTimeStatCondition;
import czopyk.model.structure.Observable;

import com.espertech.esper.client.EPAdministrator;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.collection.Pair;

public class AnalyzerSlave extends Observable {

	private static final Logger l = Logger.getLogger("AnalyzerSlave");

	private final Object ul;
	private final EPStatement statement;

	public AnalyzerSlave(EPAdministrator admin, Condition cond, final Long streamId,
			Analyzer analyzer) {

		this.addObserver(analyzer);

		// skladanie sql'a

		long round = Math.round(cond.windowSize);

		final String windowSize = cond.windowType.equals(WindowType.BATCH_LENGTH)
				|| cond.windowType.equals(WindowType.LENGTH) ? "" + round : ""
				+ cond.windowSize;// a.getRealTimeFromModelTime(c.windowSize);

		// wyrazenie na okno
		String wType = null;
		switch (cond.windowType) {
		case BATCH_LENGTH: {
			wType = "length_batch";
			break;
		}
		case LENGTH: {
			wType = "length";
			break;
		}
		case MODEL_TIME: {
			wType = "ext_timed";
		}
		}
		String window = "win:"
				+ wType
				+ "("
				+ (WindowType.MODEL_TIME.equals(cond.windowType) ? "timestamp, "
						+ windowSize + " sec" : windowSize) + ")";

		// sql
		String s = null;

		// nazwa pola w zapytanie sql, które należy odczytać
		String field = null;

		// warunki na obciazenie pojedynczych centrow
		if (cond instanceof LoadCondition) {
			LoadCondition lc = (LoadCondition) cond;

			field = "average";

			s = "select " + field + " from LoadReportEvent(centerName='"
					+ lc.centerName + "')." + window + ".stat:uni(load)";

			ul = getUl(field, streamId, analyzer, null);
			// warunki z globalnym czasem odpowiedzi
		} else if (cond instanceof ResponseTimeCondition) {
			final ResponseTimeCondition rtc = (ResponseTimeCondition) cond;

			String where = "";
			if (rtc.lc != null) {
				if (rtc.lc.getType() != null) {
					where += " loadClass.type.value = '" + rtc.lc.getType() + "' ";
					if (rtc.lc.getClazz() != null) {
						where += ", loadClass.clazz.value = '" + rtc.lc.getClazz() + "' ";
					}
				} else if (rtc.lc.getClazz() != null) {
					where += " loadClass.clazz.value = '" + rtc.lc.getClazz() + "' ";
				}
			}

			field = "time";
			String orderByAndLimitClause = " output snapshot every 1 events order by time asc";

			s = "select " + field + " from RequestCompleteEvent(" + where.trim()
					+ ")." + window + " " + orderByAndLimitClause;

			ul = getUl(field, streamId, analyzer, rtc.boundary);
		} else if (cond instanceof ResponseTimeStatCondition) {
			ResponseTimeStatCondition rt = (ResponseTimeStatCondition) cond;

			String where = "";
			if (rt.lc != null) {
				if (rt.lc.getType() != null) {
					where += " loadClass.type.value = '" + rt.lc.getType() + "' ";
					if (rt.lc.getClazz() != null) {
						where += ", loadClass.clazz.value = '" + rt.lc.getClazz() + "' ";
					}
				} else if (rt.lc.getClazz() != null) {
					where += " loadClass.clazz.value = '" + rt.lc.getClazz() + "' ";
				}
			}

			StatsType st = rt.statsType;
			if (StatsType.THROUGHPUT.equals(st)) {
				field = StatsType.DATAPOINTS.toString().toLowerCase();
				double tmp = cond.windowSize / 1000.0;
				field = "(" + field + "/" + tmp + ")";
			} else {
				field = st.toString().toLowerCase();
			}

			s = "select " + field + " from RequestCompleteEvent(" + where.trim()
					+ ")." + window + ".stat:uni(time)";

			ul = getUl(field, streamId, analyzer, null);
		} else {
			throw new IllegalArgumentException("Unknown condition class.");
		}

		l.info("Registering statement: " + s);
		statement = admin.createEPL(s);
		statement.setSubscriber(ul);
	}

	public void destroy() {
		this.deleteObservers();
	}

	@SuppressWarnings("unused")
	private Object getUl(final String field, final long streamId,
			final Analyzer a, final Double boundary) {
		if (boundary == null) {
			// pojedynczna agregacja
			return new Object() {
				public void update(double val) {
					// żaden wynik nie może być ujemny - poprawiam
					val = Math.max(0.0, val);
					val = a.cutPrecision(val);
					notifyObservers(new Pair<Long, Double>(streamId, val));
				}
			};
		} else {
			// snapshoty, wyciagania miar rozkladu
			return new Object() {
				public void update(double[] values, double[] out) {
					// żaden wynik nie może być ujemny - poprawiam

					// values jest posortowane rosnaco
					// szukam indeksu
					int index = (int) Math.round(boundary * values.length);

					// zakres
					index = Math.max(index, 0);
					index = Math.min(index, values.length - 1);

					double val = values[index];
					val = Math.max(0.0, val);
					val = a.cutPrecision(val);
					notifyObservers(new Pair<Long, Double>(streamId, val));
				}
			};
		}
	}
}
